package com.jorado.pulsar.client;

import com.jorado.pulsar.extend.util.GetAppNameUtil;
import com.jorado.pulsar.metrics.PulsarMetricsRecorder;
import com.jorado.pulsar.tracing.TracingProducerInterceptor;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.ProducerBuilderImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class ProducerCreator<T> extends ProducerBuilderImpl<T> {

    protected boolean enableMetricsCollect = true;

    protected boolean enableTracing = false;

    protected String projectName = "default";

    protected TracingProducerInterceptor<T> traceInterceptor;

    public ProducerCreator(PulsarClientImpl client, Schema<T> schema) {
        super(client, schema);
    }


    @Override
    public ProducerCreator<T> schema(Schema<T> schema) {
        super.schema(schema);
        return this;
    }

    @Override
    public ProducerCreator<T> clone() {
        throw new UnsupportedOperationException();
    }

    @Override
    public ProducerCreator<T> loadConf(Map<String, Object> config) {
        super.loadConf(config);
        return this;
    }

    @Override
    public ProducerCreator<T> topic(String topicName) {
        super.topic(topicName);
        return this;
    }

    @Override
    public ProducerCreator<T> producerName(String producerName) {
        super.producerName(producerName);
        return this;
    }

    @Override
    public ProducerCreator<T> sendTimeout(int sendTimeout, TimeUnit unit) {
        super.sendTimeout(sendTimeout, unit);
        return this;
    }

    @Override
    public ProducerCreator<T> maxPendingMessages(int maxPendingMessages) {
        super.maxPendingMessages(maxPendingMessages);
        return this;
    }

    @Override
    public ProducerCreator<T> maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) {
        super.maxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions);
        return this;
    }

    @Override
    public ProducerCreator<T> blockIfQueueFull(boolean blockIfQueueFull) {
        super.blockIfQueueFull(blockIfQueueFull);
        return this;
    }

    @Override
    public ProducerCreator<T> messageRoutingMode(MessageRoutingMode messageRouteMode) {
        super.messageRoutingMode(messageRouteMode);
        return this;
    }

    @Override
    public ProducerCreator<T> compressionType(CompressionType compressionType) {
        super.compressionType(compressionType);
        return this;
    }

    @Override
    public ProducerCreator<T> hashingScheme(HashingScheme hashingScheme) {
        super.hashingScheme(hashingScheme);
        return this;
    }

    @Override
    public ProducerCreator<T> messageRouter(MessageRouter messageRouter) {
        super.messageRouter(messageRouter);
        return this;
    }

    @Override
    public ProducerCreator<T> enableBatching(boolean batchMessagesEnabled) {
        super.enableBatching(batchMessagesEnabled);
        return this;
    }

    @Override
    public ProducerCreator<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
        super.cryptoKeyReader(cryptoKeyReader);
        return this;
    }

    @Override
    public ProducerCreator<T> addEncryptionKey(String key) {
        super.addEncryptionKey(key);
        return this;
    }

    @Override
    public ProducerCreator<T> cryptoFailureAction(ProducerCryptoFailureAction action) {
        super.cryptoFailureAction(action);
        return this;
    }

    @Override
    public ProducerCreator<T> batchingMaxPublishDelay(long batchDelay, TimeUnit timeUnit) {
        super.batchingMaxPublishDelay(batchDelay, timeUnit);
        return this;
    }

    @Override
    public ProducerCreator<T> batchingMaxMessages(int batchMessagesMaxMessagesPerBatch) {
        super.batchingMaxMessages(batchMessagesMaxMessagesPerBatch);
        return this;
    }

    @Override
    public ProducerCreator<T> initialSequenceId(long initialSequenceId) {
        super.initialSequenceId(initialSequenceId);
        return this;
    }

    @Override
    public ProducerCreator<T> property(String key, String value) {
        super.property(key, value);
        return this;
    }

    @Override
    public ProducerCreator<T> properties(Map<String, String> properties) {
        super.properties(properties);
        return this;
    }

    @Override
    public ProducerCreator<T> intercept(ProducerInterceptor<T>... interceptors) {
        super.intercept(interceptors);
        return this;
    }

    public ProducerCreator<T> metricsCollect(boolean enableMetricsCollect) {
        this.enableMetricsCollect = enableMetricsCollect;
        return this;
    }

    public ProducerCreator<T> tracing(boolean enableTracing) {
        return tracing(enableTracing, GetAppNameUtil.getAppName());
    }

    public ProducerCreator<T> tracing(boolean enableTracing, String projectName) {
        this.enableTracing = enableTracing;
        if (StringUtils.isNotBlank(projectName)) {
            this.projectName = projectName;
        }
        if (traceInterceptor == null) {
            traceInterceptor = new TracingProducerInterceptor<>(projectName);
            startTracing();
        }
        return this;
    }

    protected void startTracing() {
        if (enableTracing && traceInterceptor != null) {
            this.intercept(traceInterceptor);
        }
    }

    @Override
    public CompletableFuture<Producer<T>> createAsync() {
        CompletableFuture<Producer<T>> future = super.createAsync();
        future.whenComplete((producer, throwable) -> {
            if (enableMetricsCollect && throwable == null) {
                PulsarMetricsRecorder.getInstance().registerProducer(producer);
            }
        });
        return future;
    }

    public TracingProducerInterceptor<T> getTraceInterceptor() {
        return traceInterceptor;
    }
}
